package producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Properties;
import java.util.Random;

/**
 * Kafka 生产者类，用于生成随机学生数据并发送到 Kafka 主题
 */
public class kfkproducer {
    // 随机数生成器
    private static final Random random = new Random();


    /**
     * 主方法，程序入口
     * @param args 命令行参数
     */
    public static void main(String[] args) {
        // 1. 创建 Kafka 生产者的配置对象
        Properties properties = new Properties();

        // 2. 配置 Kafka 生产者属性
        // 设置 Kafka 服务器地址
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.244.11:9092");
        // 设置键的序列化类
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");
        // 设置值的序列化类
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");

        // 3. 创建 Kafka 生产者实例
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);

        // 4. 持续发送数据
        while (true) {
            // 生成随机学生数据
            String data = generateRandomData();

            // 创建 ProducerRecord 对象，指定主题和数据
            // 注意: 原代码中主题名是 "t2025"，但根据需求应该是 "stuInfo"
            kafkaProducer.send(new ProducerRecord<String, String>("stuInfo", data));

            // 打印发送的数据
            System.out.println("Sent data: " + data);

            try {
                // 每隔5秒推送一次
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 生成随机学生数据
     * @return 格式化后的学生数据字符串，字段用制表符分隔
     */
    private static String generateRandomData() {
        // 预定义的姓名和班级ID数组
        String[] names = {"Alice", "Bob", "Charlie", "David", "Eva"};
        String[] classIds = {"101", "102", "103", "104"};

        // 随机生成各个字段
        String classId = classIds[random.nextInt(classIds.length)];  // 随机班级
        String name = names[random.nextInt(names.length)];          // 随机姓名
        int gender = random.nextBoolean() ? 1 : 0;                 // 随机性别(1男/0女)
        String birthDate = generateRandomDate1();                   // 随机出生日期
        String studentId = generateRandomStudentId();               // 随机学号
        int semester = random.nextInt(8) + 1;                       // 随机学期(1-8)
        double gpa = 1.0 + random.nextDouble() * 9.0;               // 随机绩点(1.0-10.0)
        String enrollmentStatus = random.nextBoolean() ? "A" : "L";  // 随机在校状态(A在校/L毕业)

        // 格式化输出，字段间用制表符分隔
        return String.format("%s\t%s\t%d\t%s\t%s\t%d\t%.1f\t%s",
                classId, name, gender, birthDate, studentId, semester, gpa, enrollmentStatus);
    }

    /**
     * 生成随机日期(1970年至今)
     * @return 格式化的日期字符串(yyyy-MM-dd)
     */
    private static String generateRandomDate1() {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
        long minDate = 315569520000L; // 1970-01-01 的时间戳
        long maxDate = System.currentTimeMillis(); // 当前时间戳
        // 生成1970年至今的随机时间戳
        long randomDate = minDate + (long) (Math.random() * (maxDate - minDate));
        return sdf.format(new Date(randomDate));
    }

    /**
     * 生成随机学号
     * @return 8位数字的学号字符串
     */
    private static String generateRandomStudentId() {
        // 生成8位数字，不足前面补0
        return String.format("%08d", random.nextInt(100000000));
    }
}